/*  
  -- The Lime II Project -- 

  A tuplespaces-based middleware for coordinating agents and mobile hosts.
  Copyright (C) 2005.
  Gian Pietro Picco, Amy L. Murphy, Lorenzo Bellini.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
 */

package lime2;

import java.util.*;

/**
 * Models a buffer for receiving remote ops.
 * 
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini
 */

class IncomingBuffer {
	// private data members
	private boolean enabled; // do we want the waiting thread to be waked up by
								// the arrival of a remote op ?
	private boolean renewed; // has the last timeout been renewed ?
	private long extended; // extended time for the timeout
	private Vector buffer; // "concrete" buffer
	private RemoteOp current; // the remote op on which we are currently
								// waiting, null if none

	/**
	 * Constructor method.
	 */
	IncomingBuffer() {
		enabled = true;
		renewed = false;
		extended = 0;
		buffer = new Vector();
		current = null;
	}

	/**
	 * Adds a remote op to this buffer.
	 * 
	 * @param ro
	 *            remote op to add
	 */
	synchronized void add(RemoteOp ro) {
		buffer.addElement(ro);
		if (enabled)
			notify();
	}

	/**
	 * Renews the last timeout.
	 */
	synchronized void renew() {
		renewed = true;
	}

	/**
	 * Extends the timeout.
	 * 
	 * @param extended
	 *            the number of milliseconds the timeout has to be extended with
	 */
	synchronized void extend(long extended) {
		this.extended += extended;
	}

	/**
	 * Blocks waiting for an op that is a reply for the given remote op; removes
	 * it from the buffer and returns it. It is guaranteed that the thread will
	 * not wait longer than the timeout chosen by the user (unless the timeout
	 * is extended or renewed by the FragmentationProxy): as soon as a matching
	 * op arrives the thread is resumed and the result is returned. <br>
	 * To be used in unicast communication (.. since it checks ID of the
	 * replying host).
	 * 
	 * @param ro
	 *            given remote op to match
	 * @param timeout
	 *            maximum amount of time (milliseconds) to wait
	 * @return matching remote op, <code>null</code> otherwise
	 */
	synchronized RemoteOp remove(RemoteOp ro, long timeout) {
		current = ro;
		int rec = -1; // records the "latest" element such that previous
						// elements have already been checked
		do {
			long timestamp1 = new Date().getTime();
			long timestamp2 = timestamp1 + timeout;
			renewed = false;
			do {
				if (extended != 0) {
					timestamp2 = timestamp2 + extended;
					extended = 0;
				}
				;
				try {
					wait(timestamp2 - timestamp1);
					timestamp1 = new Date().getTime();
				} catch (InterruptedException ie) {
					ie.printStackTrace();
				}
				int i = buffer.size();
				while (--i > rec) {
					RemoteOp tmp = (RemoteOp) buffer.get(i);
					if (RemoteOp.match(ro, tmp)) {
						current = null;
						return (RemoteOp) buffer.remove(i);
					}
				}
				rec = buffer.size() - 1;
			} while (timestamp2 + extended - timestamp1 > 0);
		} while (renewed);
		current = null;
		return null;
	}

	/*
	 * If the remote op is the tree-based multicast, using this specific remove
	 * method as following
	 */
	boolean targetMatch(HashSet<LimeServerID> targets, LimeServerID replySource) {
		return targets.contains(replySource) ? true : false;
	}

	/**
	 * Remove a set of incoming operations that is replies from the specified remote 
	 * targets. If they are not arrived, block and wait for their arrivals until time
	 * out is reaching. It is guaranteed that the thread will
	 * not wait longer than the timeout chosen by the user (unless the timeout
	 * is extended or renewed by the FragmentationProxy); as soon as all matching
	 * op arrives the thread is resumed and the result is returned. <br>
	 * 
	 * When a reply arrives, its source in <code>targets</code> set will be discarded.
	 * Therefore, if this method ends with all replies arrivals, targets set will be
	 * cleared; Otherwise, the remaining items in <code>targets</code> indicate that
	 * they do not reply properly.<br>
	 * 
	 * This is to be used for OUT_MULTI
	 * @param ro the current waiting operation
	 * @param targets that hosts that replies come from
	 * @param timeout total time out for this removal
	 * @return the ArrayList object which contains all received replies
	 * @author Xing Jiankuan
	 */
	synchronized ArrayList remove(RemoteOp ro, HashSet<LimeServerID> targets,
			long timeout) {
		
		current = ro;
		int rec = -1; // records the "latest" element such that previous
						// elements have already been checked
		
		ArrayList ret = new ArrayList(); //of RemoteOp
		do {
			renewed = false;
			long timestamp1 = new Date().getTime();
			long timestamp2 = timestamp1 + timeout;
			do {
				//first check the buffer
				int i = buffer.size();
				while(--i > rec) {
					RemoteOp currOp = (RemoteOp) buffer.get(i);
					if (targetMatch(targets, currOp.source)) {
						ret.add(currOp);
						targets.remove(currOp.source);
						
					}
				}
				if (targets.isEmpty()) {
					break; //all the replies have arrived
				}
				
				rec = buffer.size() - 1;
				
				//then wait
				if (extended != 0) {
					timestamp2 += extended;
					extended = 0;
				}
				
				try {
					wait(timestamp2 - timestamp1);
					timestamp1 = new Date().getTime();
				} catch (InterruptedException ie) {
					ie.printStackTrace();
				}
			} while(timestamp2 + extended - timestamp1 > 0);
		} while (renewed);
		
		current = null;
		for(int i = 0; i < ret.size(); i++) {
			buffer.remove(ret.get(i));
		}
		return ret;
	}

	/**
	 * Multicast version of the previous method (.. the only difference is that
	 * it ignores ID of the replying host).
	 * 
	 * @param ro
	 *            given remote op to match
	 * @param timeout
	 *            maximum amount of time (milliseconds) to wait
	 * @return matching remote op, <code>null</code> otherwise
	 */
	synchronized RemoteOp removeM(RemoteOp ro, long timeout) {
		current = ro;
		int rec = -1; // records the "latest" element such that previous
						// elements have already been checked
		do {
			long timestamp1 = new Date().getTime();
			long timestamp2 = timestamp1 + timeout;
			renewed = false;
			do {
				if (extended != 0) {
					timestamp2 = timestamp2 + extended;
					extended = 0;
				}
				;
				try {
					wait(timestamp2 - timestamp1);
					timestamp1 = new Date().getTime();
				} catch (InterruptedException ie) {
					ie.printStackTrace();
				}
				int i = buffer.size();
				while (--i > rec) {
					RemoteOp tmp = (RemoteOp) buffer.get(i);
					if (ro.ti == tmp.ti && ro.seqnum + 1 == tmp.seqnum) {
						current = null;
						return (RemoteOp) buffer.remove(i);
					}
				}
				rec = buffer.size() - 1;
			} while (timestamp2 + extended - timestamp1 > 0);
		} while (renewed);
		current = null;
		return null;
	}

	/**
	 * Removes immediately a matching op from the buffer. <br>
	 * To be used in unicast communication (.. since it checks ID of the
	 * replying host).
	 * 
	 * @param ro
	 *            given remote op to match
	 * @return matching remote op, <code>null</code> otherwise
	 */
	synchronized RemoteOp remove(RemoteOp ro) {
		int i = buffer.size();
		while (--i > -1) {
			RemoteOp tmp = (RemoteOp) buffer.get(i);
			if (RemoteOp.match(ro, tmp))
				return (RemoteOp) buffer.remove(i);
		}
		return null;
	}

	/**
	 * Multicast version of the previous method (.. the only difference is that
	 * it ignores ID of the replying host).
	 * 
	 * @param ro
	 *            given remote op to match
	 * @return matching remote op, <code>null</code> otherwise
	 */
	synchronized RemoteOp removeM(RemoteOp ro) {
		int i = buffer.size();
		while (--i > -1) {
			RemoteOp tmp = (RemoteOp) buffer.get(i);
			if (ro.ti == tmp.ti && ro.seqnum + 1 == tmp.seqnum)
				return (RemoteOp) buffer.remove(i);
		}
		return null;
	}

	/**
	 * Blocks the thread for the given amount of time. Arrival of remote ops has
	 * no effect.
	 * 
	 * @param timeout
	 *            the number of milliseconds the thread must be suspended
	 */
	synchronized void timer(long timeout) {
		long bak = timeout;
		enabled = false;
		do {
			if (renewed) {
				timeout = bak;
				renewed = false;
			}
			if (extended != 0) {
				timeout += extended;
				extended = 0;
			}
			try {
				wait(timeout);
				timeout = 0;
			} catch (InterruptedException ie) {
				ie.printStackTrace();
			}
		} while (renewed || extended != 0);
		enabled = true;
	}

	/**
	 * Blocks the thread until timeout milliseconds have elapsed from the
	 * arrival of the last matching remote op; the timeout can still be extended
	 * or renewed. <br>
	 * Useful for multicast communication (.. ignores IDs of the replying hosts
	 * and sequence numbers).
	 * 
	 * @param ro
	 *            given remote op to match
	 * @param timeout
	 *            maximum amount of time (milliseconds) to wait
	 * @return matching remote op, <code>null</code> otherwise
	 */
	synchronized void expire(RemoteOp ro, long timeout) {
		current = ro;
		int rec = -1; // .. records the last element on which we "renewed"
		do {
			long timestamp1 = new Date().getTime();
			long timestamp2 = timestamp1 + timeout;
			renewed = false;
			do {
				if (extended != 0) {
					timestamp2 = timestamp2 + extended;
					extended = 0;
				}
				;
				try {
					wait(timestamp2 - timestamp1);
					timestamp1 = new Date().getTime();
				} catch (InterruptedException ie) {
					ie.printStackTrace();
				}
				boolean found = false;
				int i = buffer.size();
				while (!found && --i > rec) {
					RemoteOp tmp = (RemoteOp) buffer.get(i);
					if (ro.ti == tmp.ti) {
						renewed = true;
						rec = i;
						found = true;
					}
				}
			} while (timestamp2 + extended - timestamp1 > 0);
		} while (renewed);
		current = null;
	}

	/**
	 * Removes and returns from the buffer all the ops that match the given
	 * remote op. If no matching op is found <code>null</code> is returned. <br>
	 * Useful for multicast communication (.. ignores IDs of the replying hosts
	 * and sequence numbers).
	 * 
	 * @param ro
	 *            given remote op to match
	 * @return array of matching RemoteOps
	 */
	synchronized RemoteOp[] removeAll(RemoteOp ro) {
		Vector v = new Vector();
		int n = buffer.size();
		int i = 0;
		while (i < buffer.size()) {
			RemoteOp tmp = (RemoteOp) buffer.get(i);
			if (ro.ti == tmp.ti) {
				v.addElement(buffer.remove(i));
				n--;
			} else
				i++;
		}
		RemoteOp[] ret = null;
		if (v.size() > 0) {
			ret = new RemoteOp[v.size()];
			v.copyInto(ret);
		}
		return ret;
	}

	/**
	 * Removes all duplicated ops from the buffer. <br>
	 * Useful for multicast communication, if duplication cache is disabled.
	 * 
	 * @return this buffer
	 */
	synchronized IncomingBuffer removeDuplicates() {
		if (!LimeConstants.CACHE_DUPS_ENABLED) {
			int n = buffer.size();
			int i = 0;

			while (i < n) {
				RemoteOp tmp1 = (RemoteOp) buffer.get(i);
				int j = i + 1;
				while (j < n) {
					RemoteOp tmp2 = (RemoteOp) buffer.get(j);
					if (RemoteOp.duplicates(tmp1, tmp2)) {
						buffer.remove(j);
						n--;
					} else
						j++;
				}
				i++;
			}
		}

		return this;
	}

	/**
	 * Returns the remote op on which we are currently waiting, if any.
	 * Otherwise <code>null</code> is returned.
	 * 
	 * @return remote op or <code>null</code>
	 */
	RemoteOp getWaitingOp() {
		return current;
	}

	/**
	 * Cleans the buffer.
	 */
	synchronized void clean() {
		buffer.clear();
	}

}
